package com.wangyh8216.demo.flink.caculate;

// Flink相关导入
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

// Java标准库导入
import java.time.Duration;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Random;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class WindowRatioJob {

    /**
     * 日志模式正则表达式
     * 用于解析Apache/Nginx风格的访问日志格式
     */
    private static final Pattern LOG_PATTERN = Pattern.compile(
            "^(\\S+) - - \\[(.*?)\\] \"(.*?)\" (\\d+) (\\d+) \"(.*?)\" \"(.*?)\".*");

    // 存储上一个窗口的总访问次数
    private static long previousWindowTotal = 0L;

    public static void main(String[] args) throws Exception {

        // 1. 构造本地测试数据 - 模拟120条访问日志（两个窗口的数据）
        List<String> mockLogs = new ArrayList<>();
        // 日志模板：标准Apache/Nginx访问日志格式
        String template = "192.168.%d.%d - - [04/Sep/2025:12:00:%02d +0000] \"GET / HTTP/1.1\" 200 1234 \"-\" \"curl\"";
        
        // 生成120条测试数据，时间戳从00秒到119秒
        // IP分布：192.168.0.x 与 192.168.1.x 两个网段
        for (int i = 0; i < 120; i++) {
            int a = new Random().nextInt(2);            // 生成0或1，用于选择网段(0.x或1.x)
            int b = new Random().nextInt(3);    // 生成0-2，用于IP地址的最后一位
            mockLogs.add(String.format(template, a, b, i));  // 格式化日志并添加到列表
        }

        // 2. 创建流处理执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度为1，方便观察输出结果
        env.setParallelism(1);

        // 3. 从内存集合创建数据源（Source）
        DataStream<String> raw = env.fromCollection(mockLogs);

        // 4. 数据转换：解析日志行 -> (ip, 1L) 二元组
        SingleOutputStreamOperator<Tuple2<String, Long>> parsed = raw
                .map(new MapFunction<String, Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> map(String line) throws Exception {
                        // 使用正则表达式匹配日志行
                        Matcher m = LOG_PATTERN.matcher(line);
                        if (m.find()) {
                            // 提取IP地址（第一个分组）
                            String ip = m.group(1);
                            // 返回(IP, 1)元组，表示该IP访问一次
                            return Tuple2.of(ip, 1L);
                        }
                        // 如果日志格式不匹配，返回null
                        return null;
                    }
                })
                // 显式指定返回类型，避免Java类型擦除问题
                .returns(org.apache.flink.api.common.typeinfo.Types.TUPLE(org.apache.flink.api.common.typeinfo.Types.STRING, org.apache.flink.api.common.typeinfo.Types.LONG));

        // 5. 分配时间戳和水印（允许2秒乱序）
        SingleOutputStreamOperator<Tuple2<String, Long>> withWatermark =
                parsed.assignTimestampsAndWatermarks(
                        // 使用有界乱序水印策略，允许最多2秒的乱序
                        WatermarkStrategy.<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                                .withTimestampAssigner(new SerializableTimestampAssigner<Tuple2<String, Long>>() {
                                    private long counter = 0L;
                                    
                                    @Override
                                    public long extractTimestamp(Tuple2<String, Long> element, long recordTimestamp) {
                                        // 为每条数据分配递增的时间戳，间隔100毫秒
                                        // 120条数据分布在12秒内（跨越2个10秒窗口）
                                        return System.currentTimeMillis() - (120 * 100) + (counter++ * 100);
                                    }
                                }));

        // 6. 窗口聚合：10秒翻滚窗口统计每个窗口的总访问次数
        SingleOutputStreamOperator<Long> windowTotal = withWatermark
                // 使用windowAll对所有数据进行全局窗口处理
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(10)))
                // 使用reduce聚合函数计算每个窗口的总访问次数
                .reduce(new ReduceFunction<Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> reduce(Tuple2<String, Long> v1, Tuple2<String, Long> v2) {
                        // 累加计数，忽略IP字段
                        return Tuple2.of("total", v1.f1 + v2.f1);
                    }
                })
                // 提取总访问次数
                .map(new MapFunction<Tuple2<String, Long>, Long>() {
                    @Override
                    public Long map(Tuple2<String, Long> value) throws Exception {
                        return value.f1;
                    }
                });

        // 7. 环比计算：计算当前窗口与上一个窗口的比值
        SingleOutputStreamOperator<String> ratioResult = windowTotal
                .map(new MapFunction<Long, String>() {
                    // 使用Flink的状态变量来存储上一个窗口的值
                    private long previousWindowTotal = 0L;
                    private int windowCount = 0;
                    
                    @Override
                    public String map(Long currentTotal) throws Exception {
                        double ratio = 0.0;
                        if (previousWindowTotal > 0) {
                            // 计算环比：当前窗口/上一个窗口
                            ratio = (double) currentTotal / previousWindowTotal;
                        }
                        
                        // 保存当前窗口的总数作为下一个窗口的上一个值
                        long tempPrevious = previousWindowTotal;
                        previousWindowTotal = currentTotal;
                        
                        // 格式化输出结果，添加窗口序号和详细信息
                        windowCount++;
                        return String.format("Window %d: current=%d, previous=%d, ratio=%.2f (%s)",
                                windowCount, currentTotal, tempPrevious, ratio,
                                ratio > 1.0 ? "增长" : (ratio < 1.0 ? "下降" : "持平"));
                    }
                });

        // 8. 输出结果：将处理结果打印到控制台
        ratioResult.print();

        // 9. 执行流处理任务
        env.execute("Window Ratio Calculation");
    }
}